package com.eversec.msg.controller;


import com.eversec.kafka.consumer.KafkaConsumerServer;
import com.eversec.msg.service.KafkaService;
import com.eversec.msg.util.KafkaUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestMethod;
import org.springframework.web.bind.annotation.RestController;

import javax.annotation.PostConstruct;
import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;
import java.util.List;

//@Api(value = "kafka服务接口",produces= MediaType.APPLICATION_JSON_VALUE)
@RestController
@RequestMapping("/kafka/*")
public class CollectController {
    protected final Logger logger = LoggerFactory.getLogger(this.getClass());


    @Autowired
    private KafkaService kafkaService;

    //@ApiOperation("获取多个ip详细信息")
    @RequestMapping(value = "/send", method = RequestMethod.GET)
    public void sendKafka(HttpServletRequest request, HttpServletResponse response) {
        try {
            String message = request.getParameter("message");
            logger.info("kafka的消息={}"+ message);
            kafkaService.sendKafkaMessage("topic_miit_malice",message);
            logger.info("发送kafka成功");
        } catch (Exception e) {
            logger.error("发送kafka失败");
        }
    }


    @PostConstruct
    public void getMsg(){

        KafkaConsumerServer kafkaConsumerServer= KafkaUtil.getKafkaConsumerServerByTopic("topic_miit_malice");
        //KafkaConsumerServer kafkaConsumerServer= new KafkaConsumerServer(PropertiesInfo.KAFKA_CONSUMER_HOST,"wjbTest");
        kafkaConsumerServer.start();
        while(true){
            //调用研判报告JSON解析方法
            List<String> messages=kafkaConsumerServer.getMessages();
            for (String message : messages) {
                logger.info("消费者消息："+message);

            }
        }
    }

}
